    package com.ruirui.mainAPI;

    import reactor.core.publisher.Flux;
    import reactor.core.publisher.Mono;

    import java.time.Duration;
    import java.util.concurrent.atomic.AtomicInteger;

    public class ApiTest {


        /**
         * 找出1-5的偶数
         */
        /**
         * [ INFO] (main) | onSubscribe([Fuseable] FluxFilterFuseable.FilterFuseableSubscriber)
         * [ INFO] (main) | request(unbounded) 无限制请求
         * [ INFO] (main) | onNext(2)
         * [ INFO] (main) | onNext(4)
         * [ INFO] (main) | onComplete()
         */

        public void filter() {
          // 调出偶数
            Flux.just( 1, 2, 3, 4, 5)
                    .log()
                    .filter(e -> e % 2 == 0)//一个一个要
                    .subscribe();
        }
        public void flatMap(){
            Flux.just("zhang san","lisi","wangwu")
                    .flatMap(e -> Flux.just(e.split(" ")))
                    .log()
                    .subscribe();
        }
        public  void concatMap(){
            //流1
            Flux<String> just = Flux.just("hello", "lisi", "wangwu");
            //流2
            Flux<String> just1 = Flux.just("zhang san", "lisi", "wangwu");

            Flux.concat(just,just1)
                    .subscribe(System.out::println);
        }

        /**
         * 不会共享外部变量 无状态转换
         */
        public void transform(){
            AtomicInteger atomicInteger = new AtomicInteger(0);
            Flux<String> flux = Flux.just("hello", "lisi", "wangwu");
            Flux<String> transform = flux.transform(v -> {
                if (atomicInteger.incrementAndGet() == 1) {
                    return v.map(String::toUpperCase);
                } else {
                    return v;
                }
            });
            transform.subscribe(System.out::println);
            transform.subscribe(System.out::println);
            /**
             * | 方法                    | 一句话总结                                  |
             * | --------------------- | -------------------------------------- |
             * | `transform()`         | **共享转换逻辑**，只调用一次 lambda，所有订阅共享这份转换后的流  |
             * | `transformDeferred()` | **每个订阅者都重新执行 lambda**，拥有独立的转换逻辑（不共享状态） |
             */
        }

        /**
         *  有状态转换
         */
        public void transformDeferred(){
            AtomicInteger atomicInteger = new AtomicInteger(0);
            Flux<String> flux = Flux.just("hello", "lisi", "wangwu");
            Flux<String> transform = flux.transformDeferred(v -> {
                if (atomicInteger.incrementAndGet() == 1) {
                    return v.map(String::toUpperCase);
                } else {
                    return v;
                }
            });
            transform.subscribe(System.out::println);
            transform.subscribe(System.out::println);

//            与tranform唯一不同的是：每次 subscribe 时都会调用 lambda 表达式；
//
//            所以第一次订阅满足 atomicInteger.incrementAndGet() == 1，变大写；
//
//            第二次订阅变成普通小写；
//
//            每个订阅都是“私有的一份”，即 有状态、无共享
        }

        /**
         * 压缩
         */
        public  void zip(){
            Flux.just(1, 2, 3)
                    .zipWith(Flux.just("a", "b", "c"))
                    .log()
                    // getT1()元组的第一个元素 getT2()元组的第二个元素
                    .map(v -> v.getT1() + "==>" + v.getT2())
                    .log()
                    .subscribe(System.out::println);
        }
        public  void merge() throws Exception{
            /**
             * 与concat() 的区别  concat() 是串行执行zh，merge() 是并行执行
             */
            Flux.merge(
                    Flux.just(1, 2, 3).
                            delayElements(Duration.ofMillis(500)),
                    Flux.just(4, 5, 6)
                            .delayElements(Duration.ofMillis(1000)),
                    Flux.just("hh","nihao","www")
                            .delayElements(Duration.ofMillis(1500))
            ).subscribe(System.out::println);
            System.in.read();
        }
        public void empty(){
            // Mono.empty() 流里没元素 默认元素
            Mono.empty().defaultIfEmpty(4)
                    .subscribe(System.out::println);

//            如果原始 Mono 是空的，就“切换到另一个 Mono”；
//
//            更灵活，因为可以指定任何 Mono，不仅仅是值，也可以是异步计算、网络请求等；
//
//            语义上更像是“兜底方案”。
            Mono.empty().switchIfEmpty(Mono.just(4))
                    .subscribe(System.out::println);

        }


        public static void main(String[] args) throws Exception{

    //       new ApiTest().filter();
    //        new ApiTest().flatMap();
    //     new ApiTest().concatMap();
//           new ApiTest().transform();
//           new ApiTest().transformDeferred();
//            new ApiTest().empty();

//            new ApiTest().merge();
            new ApiTest().zip();
        }
    }
